package com.bdi.mqtt_broker.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bdi.mqtt_broker.constant.Constants;
import com.bdi.mqtt_broker.model.SubscribeStore;
import com.bdi.mqtt_broker.service.ISubscribeStoreService;
import com.bdi.mqtt_broker.service.RedisService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.*;

/**
 * @author WangNing yogehaoren@gmial.com <br>
 * @since 1.0
 *
 */
@Service
public class ISubscribeStoreServiceImpl implements ISubscribeStoreService {

    @Autowired
    RedisService redisService;

    @Override
    public void put(String topicFilter, SubscribeStore subscribeStore) {
        if(topicFilter.endsWith("#")||topicFilter.endsWith("+")){
            // 使用通配符订阅
            redisService.hset(Constants.SUB_WITH_WILDCARD_CACHE_PRE + topicFilter,
                    subscribeStore.getClientId(), SubscribeStore.toJsonString(subscribeStore) );
        }else {
            // 不使用通配符订阅
            redisService.hset(Constants.SUB_WITHOUT_WILDCARD_CACHE_PRE + topicFilter,
                    subscribeStore.getClientId(), SubscribeStore.toJsonString(subscribeStore)
                    );
        }
        redisService.sadd(Constants.SUB_CLIENT_CACHE_PRE + subscribeStore.getClientId(), topicFilter);
    }

    @Override
    public void remove(String topicFilter, String clientId) {
        if(topicFilter.endsWith("#")||topicFilter.endsWith("+")){
            redisService.hdel(Constants.SUB_WITH_WILDCARD_CACHE_PRE + topicFilter, clientId);
        }else {
            redisService.hdel(Constants.SUB_WITHOUT_WILDCARD_CACHE_PRE + topicFilter, clientId);
        }
        redisService.srem(Constants.SUB_CLIENT_CACHE_PRE + clientId, topicFilter);
    }

    @Override
    public void removeForClient(String clientId) {
        for (String topicFilter:redisService.smembers(Constants.SUB_CLIENT_CACHE_PRE + clientId)){
            redisService.hdel(Constants.SUB_WITH_WILDCARD_CACHE_PRE + topicFilter, clientId);
            redisService.hdel(Constants.SUB_WITHOUT_WILDCARD_CACHE_PRE + topicFilter, clientId);
        }
        redisService.del(Constants.SUB_CLIENT_CACHE_PRE + clientId);
    }

    @Override
    public List<SubscribeStore> search(String topic) {

        final Set<String> clientIds = new HashSet<>();
        List<SubscribeStore> result = new ArrayList<>();

        // topic 不允许使用通配符，不允许使用"/"发布消息, 必须以 '/' 开头,
        if(!StringUtils.hasText(topic) ||topic.contains("#")
                ||topic.contains("+")||!topic.startsWith("/") ||topic.length() == 1){
            return Collections.emptyList();
        }

        // 1. 计算所有能够匹配的Topic通配符
        List<String> wildcardList = new ArrayList<>(); // 使用通配符能匹配的Topic
        wildcardList.add("/#");
        StringBuilder stringBuilder = new StringBuilder();

        String[] topics = topic.substring(1).split("/");
        if(topics.length>1){
            for (int i = 0; i < topics.length-1; i++) {
                stringBuilder.append("/").append(topics[i]);
                wildcardList.add(stringBuilder.toString() + "/#");
                if(i== topics.length -2){
                    wildcardList.add(stringBuilder.toString()+"/+");
                }
            }
        }
        if(topics.length == 1){
            wildcardList.add("/+");
        }

        // 2. 计算使用通配符订阅的客户端
        wildcardList.forEach( wildcardTopic ->{

            Map<String, String> subMap = redisService.hgetAll(Constants.SUB_WITH_WILDCARD_CACHE_PRE + wildcardTopic);
            subMap.forEach((key, value)->{

                if(!clientIds.contains(key)){
                    result.add(JSONObject.toJavaObject(JSON.parseObject(value), SubscribeStore.class));
                }
                clientIds.add(key);

            });

        });


        // 3. 获取未使用通配符Topic订阅的客户端
        redisService.hgetAll(Constants.SUB_WITHOUT_WILDCARD_CACHE_PRE + topic).forEach((key,value)->{

            if(!clientIds.contains(key)){
                result.add(JSONObject.toJavaObject(JSON.parseObject(value), SubscribeStore.class));
            }
            clientIds.add(key);

        });


        return result;
    }
}
